package cn.doitedu.api;

import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;

public class _02_MinMinByMaxMaxBy {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        Tuple3<String,Integer,String> p1 = new Tuple3<>("a", 28, "male");
        Tuple3<String,Integer,String> p2 = new Tuple3<>("b", 38, "male");
        Tuple3<String,Integer,String> p3 = new Tuple3<>("c", 18, "male");
        Tuple3<String,Integer,String> p4 = new Tuple3<>("x", 26, "female");
        Tuple3<String,Integer,String> p5 = new Tuple3<>("y", 16, "female");
        Tuple3<String,Integer,String> p6 = new Tuple3<>("z", 46, "female");


        ArrayList<Tuple3<String, Integer, String>> list = new ArrayList<>();
        list.add(p1);
        list.add(p2);
        list.add(p3);
        list.add(p4);
        list.add(p5);
        list.add(p6);


        DataStreamSource<Tuple3<String, Integer, String>> stream = env.fromCollection(list);


        stream.keyBy(new KeySelector<Tuple3<String, Integer, String>, String>() {
                    @Override
                    public String getKey(Tuple3<String, Integer, String> value) throws Exception {
                        return value.f2;
                    }
                })
                //.max("f1")  // 求出来的结果，只更新max目标字段，别的字段会沿用上一条结果
                .maxBy("f1")  // 求出来的结果，所有字段都用最大值所在的这一条
                .print();


        env.execute();


    }


}

